package net.message.processor;

import net.StoppableWorker;
import net.IncomingMessage;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class MessageWorker extends StoppableWorker {

    private final IMessageProcessor processor;
    private final BlockingQueue<IncomingMessage> queue = new LinkedBlockingQueue<IncomingMessage>();

    public MessageWorker(IMessageProcessor processor) {
        this.processor = processor;
    }

    public void addMessage(IncomingMessage im) {
        try {
            queue.put(im);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    protected void work() {
        try {
            IncomingMessage im = queue.take();
            im.message.accept(processor, im.sourceAddress);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

}
